package com.atguigu.flink.streamapi.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

/**
 * Created by Smexy on 2023/1/30
 *      写出的数据应该K-V结构。
 *             第一种： 不需要指定key
 *                          生产者默认用轮询的方式分区，保证负载均衡！
 *             第二种:  需要指定key
 *                          key用来分区，key相同，一定写入kafka的同一个分区！
 *
 *     -----------------
 *      本质就是构造一个生产者，在flink中称为 FlinkKafkaProducer
 */
public class Demo2_KafkaSink1
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
            第一种： 不需要指定key
            构造FlinkKafkaProducer
                 String brokerList,
                 String topicId,
                  SerializationSchema<IN> serializationSchema  序列化器( 指定value的序列化器)
                        SimpleStringSchema: 将String类型序列化
         */
        FlinkKafkaProducer kafkaProducer = new FlinkKafkaProducer("hadoop103:9092", "topicA", new SimpleStringSchema());

        env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .map(ws -> JSON.toJSONString(ws))
                   .addSink(kafkaProducer);


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
